package com.alibaba.middleware.race.rpc.netty;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.Map;
import java.util.concurrent.Semaphore;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import com.alibaba.middleware.race.rpc.async.ResponseCallbackListener;
import com.alibaba.middleware.race.rpc.async.ResponseFuture;
import com.alibaba.middleware.race.rpc.demo.service.RaceDO;
import com.alibaba.middleware.race.rpc.model.InvokeFuture;
import com.alibaba.middleware.race.rpc.model.RpcRequest;
import com.alibaba.middleware.race.rpc.model.RpcResponse;
import com.alibaba.middleware.race.rpc.serializer.RpcDecoder;
import com.alibaba.middleware.race.rpc.serializer.RpcEncoder;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timer;


public class NettyRpcConnection extends ChannelInboundHandlerAdapter implements RpcConnection {

	
	private InetSocketAddress inetAddr;
	
	private volatile Channel channel;
	
	private volatile RpcResponse response;
	
	private volatile Throwable exception;
	
	//HashedWheelTimer
	private volatile Timer timer;

	private boolean connected;
	
	private InvokeFuture<RpcResponse> future;
	
	private volatile Semaphore semp =new Semaphore(0);
	
	
	private Map<String, ResponseCallbackListener> listenermap;
	
	
	private String requestMethod;
	
	public NettyRpcConnection(String host, int port) {
		this.inetAddr = new InetSocketAddress(host, port);
		this.timer = new HashedWheelTimer();
	}
	@Override
	public void clear()
	{
//		semp =new Semaphore(0);
//		future=null;
//		response=null;
//		exception=null;
	}
	
	public void notifyListenerResponse()
	{
		if(listenermap.containsKey(requestMethod)&&listenermap.get(requestMethod)!=null)
		{
			listenermap.get(requestMethod).onResponse(response.getResult());
		}
	}
	public void notifyListenerException()
	{
		if(listenermap.containsKey(requestMethod)&&listenermap.get(requestMethod)!=null)
		{
			listenermap.get(requestMethod).onResponse(exception);
		}
	}
	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		// TODO Auto-generated method stub
		super.channelActive(ctx);
		System.out.println("active");
	}

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg)
			throws Exception {
		// TODO Auto-generated method stub
		response = (RpcResponse)msg;
		if(future!=null)
			future.setResult(response);
		
		synchronized (channel) {
			semp.release();
			notifyListenerResponse();
		}
	}

	@Override
	public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
		// TODO Auto-generated method stub
		super.channelReadComplete(ctx);
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
			throws Exception {
		// TODO Auto-generated method stub
		this.exception=cause;
		notifyListenerException();
		cause.printStackTrace();
	}

	@Override
	public RpcResponse sendRequest(RpcRequest request) throws Throwable {
		// TODO Auto-generated method stub
		if (!isConnected()) {
			throw new IllegalStateException("not connected");
		}
		ChannelFuture writeFuture = channel.writeAndFlush(request);
		if (!writeFuture.awaitUninterruptibly().isSuccess()) {
			close();
			throw writeFuture.cause();
		}
		//等待服务器结果
		waitForResponse();
		Throwable ex = exception;
		RpcResponse resp = this.response;
		this.response = null;
		this.exception = null;
		if (null != ex) {
			close();
			ex.printStackTrace();
			throw ex;
		}
		return resp;
	}

	@Override
	public void connection() throws Throwable {
		// TODO Auto-generated method stub
		if (connected) {
			return;
		}
		final NettyRpcConnection handle=this;
		EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group).channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel channel) throws Exception {
                        channel.pipeline()
                            .addLast(new RpcEncoder(RpcRequest.class))
                            .addLast(new RpcDecoder(RpcResponse.class))
                            .addLast(handle);
                        
                    }
                })
                .option(ChannelOption.SO_KEEPALIVE, true);

        ChannelFuture future = bootstrap.connect(this.inetAddr).sync();
        if (!future.awaitUninterruptibly().isSuccess()) {
			throw future.cause();
		}
		channel = future.channel();
		connected = true;
        }
        catch (Exception ex) 
        {
            //group.shutdownGracefully();
        	ex.printStackTrace();
        }
	}

	@Override
	public void close() throws Throwable {
		// TODO Auto-generated method stub
		connected = false;
		if (null != timer) {
			timer.stop();
			timer = null;
		}
		if (null != channel) {
			channel.closeFuture().sync();
			this.exception = new IOException("connection closed");
			synchronized (channel) {
				channel.notifyAll();
			}
			channel = null;
		}
		System.out.println("channel close");
		

	}
	public void waitForResponse() {
		synchronized (channel) {
			try {
				channel.wait();
			} catch (InterruptedException e) {
			}
		}
	}
	@Override
	public boolean isConnected() {
		// TODO Auto-generated method stub
		return connected;
	}

	@Override
	public boolean isClosed() {
		// TODO Auto-generated method stub
		return (null == channel) || !channel.isOpen()
				|| !channel.isWritable() || !channel.isActive();
	}


	@Override
	public synchronized RpcResponse sendRequest(RpcRequest request,
			Map<String, ResponseCallbackListener> map) throws Throwable {
		// TODO Auto-generated method stub
		listenermap=map;
		requestMethod=request.getMethodName();
		if (!isConnected()) {
			throw new IllegalStateException("not connected");
		}
		//ChannelFuture writeFuture = channel.writeAndFlush(request).sync();
		future=new InvokeFuture<RpcResponse>();
		ChannelFuture writeFuture = channel.writeAndFlush(request);
		if (!writeFuture.awaitUninterruptibly().isSuccess()) {
			close();
			throw writeFuture.cause();
		}
		try 
		{
			if(map.containsKey(request.getMethodName()))//异步的方法的话就直接返回
			{
				new ResponseFuture().setFuture(future);
				return null;
			}
		}
		finally
		{
			semp.acquire();
		}
		Throwable ex = exception;
		RpcResponse resp = this.response;
		this.response = null;
		this.exception = null;

		if (null != ex) {
			close();
			throw ex;
		}
		return resp;
	}

}
